﻿#include "UdpController.h"

#include <qnetworkdatagram.h>
#include <qdatetime.h>

#include "UdpCodec.h"
#include "MsgIdManager.h"
#include "Mapper_Test_RTT.h"
#include "RegisterCenter_Device.h"
#include "Device.h"
#include "DataPackage_UdpCodec.h"
#include "Mapper_testDataByTime.h"

UdpController::UdpController()
{
  //网络不平稳这里直接采用4rtt周期作为大周期
  cwndGainMap.insert(0,1.25);
  cwndGainMap.insert(1,0.75);
  cwndGainMap.insert(2,1);
  cwndGainMap.insert(3,1);
  pacingGainMap.insert(0,1.25);
  pacingGainMap.insert(1,0.75);
  pacingGainMap.insert(2,1);
  pacingGainMap.insert(3,1);
  connect(&test,&QTimer::timeout,this,&UdpController::slot_sendTestData);
  connect(&timer_rttCycle,&QTimer::timeout,this,&UdpController::slot_rttCycleEnd);
  connect(&timer_dataByTime,&QTimer::timeout,this,&UdpController::slot_dataByTime);
}

UdpController::~UdpController()
{
}

void UdpController::send(DataPackage_UdpCodec* dataPackage,bool emergency)
{
  if(emergency||!active_bbr)//如果紧急发送或者没有开启BBR控制
    {
      if(active_rto)
        {
          //开启重传，禁用BBR，记录重传信息
          addToInflightList(dataPackage);
          sendDataPackage(dataPackage);
        }
      else
        {
          //禁用重传，禁用BBR，直接发送，发送后删除数据包
          sendDataPackage(dataPackage);
          dataPackage->deleteLater();
        }
    }
  else
    {
      //正常发送交由发送队列处理
      dataPackageListBuffer.append(dataPackage);//先添加到缓冲区
      trySend();//尝试发送数据
    }
}

bool UdpController::receive(DataPackage_UdpCodec* dataPackage_UdpCodec)
{
  //该dataPackage_UdpCodec会在UdpCode中删除，这里不需要对其进行内存管理
  ushort timeStamp=dataPackage_UdpCodec->getTimeStamp();
  uint hashValue=dataPackage_UdpCodec->hashValue();
  if(!deliveredRecordMap.contains(timeStamp))
    {
      //校验数据合法性，检查数据包中的时间戳是否在已传输数据记录中
      //如果计算delivered的map不包含该时间戳，应该是超时了
      //超过该限制的数据包直接丢弃
      return false;
    }
  if(rttCycleList.contains(timeStamp))
    {
      //更新rtt周期
      //若时间戳属于当前rtt周期内，则结束当前rtt周期
      slot_rttCycleEnd();
    }
  //即使关闭BBR控制也要计算相关参数
  updateTcpAndBbr(dataPackage_UdpCodec);//更新TCP基本参数
  updateTimeoutTimer();//根据新的RTO更新全部超时计时器
  if(!inflightDataPackage.contains(hashValue))
    {
      if(!active_rto)
        {
          //如果关闭重传且该数据包不存在于传输中队列
          //则说明该数据包是紧急包或者关闭了BBR控制
          //交由上层处理
          return true;
        }
      else
        {
          //开启了重传机制，但数据包不存在于传输中队列
          //说明该数据包已经被ack了或者超时了，直接丢弃
          return false;
        }
    }
  else
    {
      //数据包被ack，关闭超时计时器，只要是存在与传输中队列的必有超时计时器
      QTimer *timer=inflightTimer.take(hashValue);
      timer->stop();
      timer->deleteLater();
      //数据包被ack，记录参数
      mapper_Test_RTT->record(dataPackage_UdpCodec->getMsgId(),rtt,test.interval(),bw,rtt_ave>>3,rto,bw_ave>>3);
      if(active_bbr)
        {
          //如果开启了bbr控制，则更新传输中数据包大小
          inflight -= dataPackage_UdpCodec->getData().size()+DataPackage_UdpCodec::dataHeadSize;
        }
       inflightDataPackage.take(hashValue)->deleteLater();//清除传输队列的记录
    }
  trySend();//接收到数据后尝试发送数据
  return true;
}

bool UdpController::isCongestion()
{
  return dataPackageListBuffer.size()>bufferLimit;
}

bool UdpController::isActive_bbr()
{
  return active_bbr;
}

bool UdpController::isActive_rto()
{
  return active_rto;
}

void UdpController::slot_sendTestData()
{
  udpCodec->code(100000,msgId++,UdpCodec::Test,QByteArray(dataSize,1));
}

void UdpController::slot_deviceOnline()
{
  test.start(sendInterval);
  timer_dataByTime.start(1000);
}

void UdpController::slot_deviceOffline()
{
  test.stop();
  timer_dataByTime.stop();
  for(DataPackage_UdpCodec* temp:dataPackageListBuffer)
    {
      temp->deleteLater();
    }
  dataPackageListBuffer.clear();
  for(DataPackage_UdpCodec* temp:inflightDataPackage.values())
    {
      temp->deleteLater();
    }
  inflightDataPackage.clear();
  rtt_ave=rtt_init<<3;
  rtt_mdev=rtt_init<<1;
  rtt_maxMdev=rtt_mdev;
  rtt_var=rtt_mdev;
  rto=rtt_init+rtt_mdev;
  currentRttCycle_TCP=0;
  //BBR
  rtt_min=rtt_ave;
  bw_ave=bw_init<<3;
  bdp=0;
  inflight=0;
  delivered=0;
  delivered_time=0;
  //record
  delivered_record = delivered;

  for(DeliveredRecord* temp:deliveredRecordMap.values())
    {
      temp->deleteLater();
    }
  deliveredRecordMap.clear();
  msgId=0;

  //RTT cycle
  timer_rttCycle.stop();
  rttCycleList.clear();//记录发送时间，用于rtt周期的结束
  //增益队列
  cwndGainIndex=0;
  pacingGainIndex=0;
}

void UdpController::slot_rttCycleEnd()
{
  timer_rttCycle.stop();
  rttCycleList.clear();
  cwndGainIndex=(cwndGainIndex+1)&0x03;
  pacingGainIndex=(pacingGainIndex+1)&0x03;
  if(0==cwndGainIndex)
    {
      //新的周期开始了,将平均带宽减小四分之一
      bw_ave-=(bw_ave>>2);
    }
  timer_rttCycle.start(rto);
}

void UdpController::slot_dataPackageTimeout(uint hashValue)
{
  //只有添加到传输中队列的数据包才会有超时处理，这里不用判断是否开启bbr，决定超时策略的是是否开启rto
  //数据包超时，前方传输路径可能出现拥塞或者断线
  QTimer* timer=inflightTimer.value(hashValue);
  timer->stop();
  DataPackage_UdpCodec* dataPackage = inflightDataPackage.value(hashValue);//取出对应的指令
  if(active_rto)
    {
      //开启超时重传时，超时处理为重发
      if(registerCenter_Device->isOnline(dataPackage->getDeviceId()))
        {
          //如果设备在线则重发
          sendDataPackage(dataPackage);
          timer->start(rto);
          qDebug()<<"resend data";
          return;
        }
    }
  else
    {
      //禁用超时重传时，超时处理为删除对应记录并减少inflight数据量
      inflight-=(dataPackage->getData().size()+DataPackage_UdpCodec::dataHeadSize);
    }
  //删除记录
  inflightDataPackage.take(hashValue)->deleteLater();//移除等待缓冲区中的对应记录
  inflightTimer.take(hashValue)->deleteLater();//移除超时计时器中对应的记录
}

void UdpController::slot_dataByTime()
{
  mapper_testDataByTime->record(rtt,rtt_ave>>3,delivered-delivered_record,bw_ave>>3,rto);
  delivered_record = delivered;
}

void UdpController::sendDataPackage(DataPackage_UdpCodec* dataPackage_UdpCodec)
{
  ushort timeStamp=QDateTime::currentMSecsSinceEpoch();
  dataPackage_UdpCodec->setTimeStamp(timeStamp);//设置数据包中的时间戳
  if(!deliveredRecordMap.contains(timeStamp))
    {
      //如果已发送记录不包含该时间戳，则添加该时间戳截止时的已接收记录
      DeliveredRecord* deliveredRecord=new DeliveredRecord();
      deliveredRecord->setTimeStamp(timeStamp)
          ->setDelivered(delivered)
          ->setTime(delivered_time)
          ->setUdpController(this)
          ->startAutoDelete(40*1000);
      deliveredRecordMap.insert(timeStamp,deliveredRecord);
    }
  rttCycleList.append(timeStamp);//添加当前时间到RTT周期计算队列中
  if(!timer_rttCycle.isActive())
    {
      timer_rttCycle.start(rto);
    }
  udpCodec->code(dataPackage_UdpCodec);
}

void UdpController::trySend()
{
  //先计算BDP
  //如果发送列表不为空，则持续发送，直到超过拥塞限制
  int gainedBdp=(rtt_ave>>3)*(bw_ave>>3)/1000*cwndGainMap.value(cwndGainIndex);
  int currentIndex=0;//用于规避相同哈希值的数据包
  while (gainedBdp>inflight)
    {
      if(dataPackageListBuffer.size()==currentIndex)
        {
          return;//到达队尾，直接返回
        }
      //判断是否发送缓冲区存在相同hashvalue的指令
      //从发送队列得到要发送的数据，同时在发送队列删除该条记录
      DataPackage_UdpCodec* dataPackage = dataPackageListBuffer.value(currentIndex);
      uint hashValue = dataPackage->hashValue();//计算hash value
      if(inflightDataPackage.contains(hashValue))
        {
          //如果传输中的数据包含该hashValue，则取下一个数据包发送，直到到达队列尾部
          ++currentIndex;
          continue;
        }
      //如果不在等待缓冲区中，说明可以发送该数据
      //决定要发送后
      dataPackageListBuffer.removeAt(currentIndex);//从发送缓冲队列中移除
      addToInflightList(dataPackage);//添加到传输中队列
      sendDataPackage(dataPackage);
      inflight+=(dataPackage->getData().size()+DataPackage_UdpCodec::dataHeadSize);//更新传输中的数据量
    }
}

void UdpController::addToInflightList(DataPackage_UdpCodec* dataPackage_UdpCodec)
{
  int hashValue = dataPackage_UdpCodec->hashValue();
  inflightDataPackage.insert(hashValue,dataPackage_UdpCodec);//添加到等待缓冲区
  qDebug()<<"inflightDataPackage insert="<<hashValue;
  qDebug()<<"inflightDataPackage valuse="<<inflightDataPackage.keys();
  //------------------------设置超时计时器---------------------------
  QTimer* timer=new QTimer();
  connect(timer,&QTimer::timeout,[=](){slot_dataPackageTimeout(hashValue);});
  inflightTimer.insert(hashValue,timer);//添加等待缓冲区的超时计时器
  timer->start(rto);//开启删除等待缓冲区中超时记录的计时器
  //--------------------------------------------------------------
}

void UdpController::updateTcpAndBbr(DataPackage_UdpCodec* dataPackage_UdpCodec)
{
  //更新基本TCP参数
  ushort timeStamp = dataPackage_UdpCodec->getTimeStamp();
  ushort now = delivered_time = QDateTime::currentMSecsSinceEpoch()&0xff;//得到当前时间//更新已收到数据包截止时间
  ushort rtt = this->rtt = now - timeStamp;//计算该数据包的RTT
  updateRto(rtt);//更新rto和rtt_ave不需要区分是否存在等待ack
  delivered += dataPackage_UdpCodec->getData().size()+DataPackage_UdpCodec::dataHeadSize;//更新已发送数据包大小
  DeliveredRecord *deliveredRecord = deliveredRecordMap.value(timeStamp);
  ushort temp=(now - deliveredRecord->time);
  if(0==temp)
    {
      //防止除0出错
      return;
    }
  bw = (delivered - deliveredRecord->delivered)*1000 / temp;
  bw_ave+=bw-(bw_ave>>3);
}

void UdpController::updateRto(int rtt)
{
  rtt-=(rtt_ave>>3);
  rtt_ave+=rtt;//rtt_ave=7/8*rtt_ave+1/8*rtt;
  if(0<rtt)
    {
      rtt-=(rtt_mdev>>2);
    }
  else//rtt<0
    {
      rtt=-rtt;
      rtt-=(rtt_mdev>>2);
      if(0<rtt)
        {
          rtt>>=3;
        }
    }
  rtt_mdev+=rtt;
  if(rtt_mdev>rtt_maxMdev)
    {
      rtt_maxMdev=rtt_mdev;
      if (rtt_maxMdev > rtt_var )
        rtt_var = rtt_maxMdev;
    }
  if(currentRttCycle_TCP!=cwndGainIndex)
    {
      //tcp记录的当前rtt周期不等于目前的rtt周期编号，说明rtt周期已经结束
      currentRttCycle_TCP=cwndGainIndex;
      if(rtt_maxMdev<rtt_var)
        {
//          rtt_var-=(rtt_var-rtt_maxMdev)>>2;//原始的
          rtt_var-=(rtt_var-rtt_maxMdev)>>1;
        }
      rtt_maxMdev=rto_min;
    }
  rto=(rtt_ave>>3)+rtt_var;
}

void UdpController::updateTimeoutTimer()
{
  for(uint key : inflightTimer.keys())
    {
      //rto变大,前方可能发生拥塞,计时器增加增大的数值
      //rto变小，拥塞解除,计时器减少变小的数值，若变小的数值大于剩余时间则直接触发
      QTimer* temp=inflightTimer.value(key);
      qDebug()<<"key:"<<key<<"remain time:"<<temp->remainingTime();
      int newTime=temp->remainingTime()-temp->interval()+rto;
      temp->start(newTime>0?newTime:0);
    }
}

UdpCodec* UdpController::getUdpCodec()const
{
  return udpCodec;
}

UdpController* UdpController::setUdpCodec(UdpCodec*const udpCodec)
{
  this->udpCodec=udpCodec;
  printInjection(this,udpCodec);
  return this;
}

Mapper_Test_RTT* UdpController::getMapper_Test_RTT()const
{
  return mapper_Test_RTT;
}

UdpController* UdpController::setMapper_Test_RTT(Mapper_Test_RTT*const mapper_Test_RTT)
{
  this->mapper_Test_RTT=mapper_Test_RTT;
  printInjection(this,mapper_Test_RTT);
  return this;
}

RegisterCenter_Device* UdpController::getRegisterCenter_Device()const
{
  return registerCenter_Device;
}

UdpController* UdpController::setRegisterCenter_Device(RegisterCenter_Device*const registerCenter_Device)
{
  if(Q_NULLPTR!=this->registerCenter_Device)
    {
      disconnect(registerCenter_Device,&RegisterCenter_Device::signal_deviceOnline,this,&UdpController::slot_deviceOnline);
      disconnect(registerCenter_Device,&RegisterCenter_Device::signal_deviceOffline,this,&UdpController::slot_deviceOffline);
    }
  this->registerCenter_Device=registerCenter_Device;
  printInjection(this,registerCenter_Device);
  if(Q_NULLPTR!=this->registerCenter_Device)
    {
      connect(registerCenter_Device,&RegisterCenter_Device::signal_deviceOnline,this,&UdpController::slot_deviceOnline);
      connect(registerCenter_Device,&RegisterCenter_Device::signal_deviceOffline,this,&UdpController::slot_deviceOffline);
    }
  return this;
}

Mapper_testDataByTime* UdpController::getMapper_testDataByTime()const
{
  return mapper_testDataByTime;
}

UdpController* UdpController::setMapper_testDataByTime(Mapper_testDataByTime*const mapper_testDataByTime)
{
  this->mapper_testDataByTime=mapper_testDataByTime;
  printInjection(this,mapper_testDataByTime);
  return this;
}

void UdpController::objectAutoInjection(QHash<QString, QObject *> *objMap)
{
  setUdpCodec(getObjFromMap<UdpCodec>(objMap));
  setMapper_Test_RTT(getObjFromMap<Mapper_Test_RTT>(objMap));
  setRegisterCenter_Device(getObjFromMap<RegisterCenter_Device>(objMap));
  setMapper_testDataByTime(getObjFromMap<Mapper_testDataByTime>(objMap));
}
